package com.voice.core.connection.impl;

import com.alibaba.fastjson.JSONObject;
import com.voice.core.entity.HttpHeader;
import com.voice.core.channel.IOContext;
import com.voice.core.channel.tcp.impl.SocketChannelAdapter;
import com.voice.core.connection.ConnectionStatus;
import com.voice.core.connection.callback.StreamListener;
import com.voice.core.connection.frame.*;
import lombok.Data;

import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.SocketChannel;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

@Data
public class SocketChannelConnectionImpl extends AbstractConnection {

    protected SocketChannelAdapter socketChannelAdapter;

    public SocketChannelConnectionImpl(int currentStreamId){
        super(currentStreamId);
    }

    public SocketChannelConnectionImpl(SocketChannel socketChannel, IOContext ioContext, int currentStreamId) {
        super(currentStreamId);
        this.socketChannelAdapter = new SocketChannelAdapter(ioContext, socketChannel, this);
        setStatus(ConnectionStatus.CREATING);
    }

    @Override
    public void close() {
        this.socketChannelAdapter.close();
    }

    @Override
    public void onDataRead(int streamId, byte[] bytes, boolean endOfStream) {
        Optional<StreamListener> optionalStreamListener = streamListener(streamId);
        optionalStreamListener.ifPresent(streamListener -> streamListener.onDataRead(this, stream(streamId), bytes, endOfStream));
    }

    @Override
    public void onStreamRead(int streamId, byte[] bytes) {
        if (!streamListener(streamId).isPresent()) {
            defaultStreamListener().ifPresent(streamListener -> setStreamListener(streamId, streamListener));
        }
        if (bytes != null && bytes.length > 0) {
            Optional<StreamListener> optionalStreamListener = streamListener(streamId);
            optionalStreamListener.ifPresent(streamListener -> streamListener.onStreamRead(this, bytes));
        }
    }

    @Override
    public void onHeadersRead(int streamId, byte[] bytes, byte weight, boolean endOfStream) {
        if (!streamListener(streamId).isPresent()) {
            defaultStreamListener().ifPresent(streamListener -> setStreamListener(streamId, streamListener));
        }
        try {
            HttpHeader requestHeader = JSONObject.parseObject(new String(bytes).replaceAll("[\\x00-\\x09\\x11\\x12\\x14-\\x1F\\x7F]", ""), HttpHeader.class);
            streamListener(streamId).ifPresent(streamListener -> streamListener.onHeadersRead(this, stream(streamId), requestHeader, endOfStream));
        } catch (Exception e) {
            e.printStackTrace();
            streamListener(streamId).ifPresent(streamListener -> streamListener.onStreamError(this, stream(streamId), new IOException("parse request header error, hex dump: " + bytes)));
        }
    }

    @Override
    public void onPingRead(long data) {
        int streamId = this.incrementAndGetNextStreamId();
        PingFrame pingFrame=new PingFrame(streamId,true,data);
        this.socketChannelAdapter.writeFrameAsync(pingFrame);
    }

    @Override
    public void onPingAckRead(long data) { }

    @Override
    public void onGoAwayRead(int streamId, int errorCode, byte[] bytes) {
        this.close();
    }

    @Override
    public void onUnknownFrame(int streamId, byte frameType, byte[] bytes) {
        streamListener(streamId).ifPresent(streamListener -> streamListener.onStreamError(this, stream(streamId), new IOException("unknown frame received, hex dump: " + bytes)));
    }

    private CompletableFuture<Integer> writeHeaders(HeaderFrame headerFrame,StreamListener streamListener){
        CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
        if (streamListener != null) this.setStreamListener(headerFrame.streamId, streamListener);
        this.socketChannelAdapter.writeFrameAsync(headerFrame);
        completableFuture.complete(headerFrame.streamId);
        return completableFuture;
    }

    @Override
    public CompletableFuture<Integer> writeHeaders(int streamId, byte[] bytes, boolean endStream, StreamListener streamListener) {
        HeaderFrame headerFrame = new HeaderFrame(bytes, streamId, endStream);
        return writeHeaders(headerFrame,streamListener);
    }

    @Override
    public CompletableFuture<Integer> writeHeaders(byte[] bytes, boolean endStream, StreamListener streamListener) {
        int streamId = this.incrementAndGetNextStreamId();
        HeaderFrame headerFrame = new HeaderFrame(bytes, streamId, endStream);
        return writeHeaders(headerFrame,streamListener);
    }

    @Override
    public CompletableFuture<Integer> writeHeaders(int streamId, byte[] bytes, boolean endStream, StreamListener streamListener, Runnable writeCallback) {
        HeaderFrame headerFrame = new HeaderFrame(bytes, streamId, endStream);
        headerFrame.setWriteCallback(writeCallback);
        return writeHeaders(headerFrame,streamListener);
    }

    @Override
    public CompletableFuture<Integer> writeHeaders(byte[] bytes, boolean endStream, StreamListener streamListener, Runnable writeCallback) {
        int streamId = this.incrementAndGetNextStreamId();
        HeaderFrame headerFrame = new HeaderFrame(bytes, streamId, endStream);
        headerFrame.setWriteCallback(writeCallback);
        return writeHeaders(headerFrame,streamListener);
    }

    private CompletableFuture<Integer> writeData(DataFrame dataFrame){
        CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
        this.socketChannelAdapter.writeFrameAsync(dataFrame);
        completableFuture.complete(dataFrame.streamId);
        return completableFuture;
    }
    @Override
    public CompletableFuture<Integer> writeData(int streamId, byte[] bytes, boolean endStream) {
        DataFrame dataFrame = new DataFrame(bytes, streamId, endStream);
        return writeData(dataFrame);
    }

    @Override
    public CompletableFuture<Integer> writeData(int streamId, byte[] bytes, boolean endStream, Runnable writeCallback) {
        DataFrame dataFrame = new DataFrame(bytes, streamId, endStream);
        dataFrame.setWriteCallback(writeCallback);
        return writeData(dataFrame);
    }

    @Override
    public void writePing(long data) {
        int streamId = this.incrementAndGetNextStreamId();
        PingFrame pingFrame=new PingFrame(streamId,false,data);
        this.socketChannelAdapter.writeFrameAsync(pingFrame);
    }

    @Override
    public void writeStream(InputStream inputStream) {
        int streamId = this.incrementAndGetNextStreamId();
        StreamFrame streamFrame = new StreamFrame(streamId, inputStream);
        this.socketChannelAdapter.writeFrameAsync(streamFrame);
    }


    private CompletableFuture<Integer> writeGoAway(GoAwayFrame goAwayFrame){
        CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
        this.socketChannelAdapter.writeFrameAsync(goAwayFrame);
        completableFuture.complete(goAwayFrame.streamId);
        return completableFuture;
    }
    @Override
    public CompletableFuture<Integer> writeGoAway(int errorCode, byte[] debugData) {
        int streamId = this.incrementAndGetNextStreamId();
        GoAwayFrame goAwayFrame = new GoAwayFrame(streamId, errorCode, debugData);
        return writeGoAway(goAwayFrame);
    }

    @Override
    public CompletableFuture<Integer> writeGoAway(int errorCode, byte[] debugData, Runnable writeCallback) {
        int streamId = this.incrementAndGetNextStreamId();
        GoAwayFrame goAwayFrame = new GoAwayFrame(streamId, errorCode, debugData);
        goAwayFrame.setWriteCallback(writeCallback);
        return writeGoAway(goAwayFrame);
    }
}
